#include "helper/rocksdb/Rocksdb.h"

#include <cassert>
#include <filesystem>
#include <iostream> // for cerr

#include "rocksdb/db.h"
#include "helper/rocksdb/Table.h"
#include "helper/rocksdb/Options.h"

namespace helper {
namespace rocksdb {

Rocksdb::Rocksdb(const std::string & name, ::rocksdb::Comparator * comparator)
: _name(name)
, _options(new Options(comparator))
{
}

Rocksdb::~Rocksdb() {
  this->close();

  delete _options; _options = nullptr;
}

bool Rocksdb::exist(const std::string & dbname) {
  return std::filesystem::exists(dbname);
}
bool Rocksdb::touch(const std::string & dbname, ::rocksdb::Options & options){
  ::rocksdb::DB* rdb;
  options.create_if_missing = true;
  auto s = ::rocksdb::DB::Open(options, dbname, &rdb);
  if(s.ok()){
    rdb->Close(); 
    delete rdb;
    return true;
  }
  std::cerr << "touch db failed:" << s.ToString() << std::endl;
  return false;
}

bool Rocksdb::isOpened() const{
  return _isOpened;
}


bool Rocksdb::createIfMissing() {
  if(!_isOpened){
    if(!Rocksdb::exist(_name)){
      if(false == touch(_name, _options->options())){
        return false;
      }
    }
    if(_openExist()){ //open all Column Families ,save <name, handle>=>tables
      _isOpened = true;
      return true;
    }
    return false;
  }
  return true;
}
//open exist  
//open all column families
bool Rocksdb::openExist() {
  if(!_isOpened){
    assert(Rocksdb::exist(_name)); 
    if(_openExist()){ //open all Column Families ,save <name, handle>=>tables
      _isOpened = true;
      return true;
    }
    return false;
  }
  return true;
}
void Rocksdb::close() {
  if(_isOpened){
        for(auto && [name, handle] : _tables){
            this->_closeColumnFamilyHandle(handle); 
        }
        _tables.clear();

    _db->Close();
    _db.reset();
    _isOpened = false;
  }
}

std::pair<bool, ::rocksdb::ColumnFamilyHandle*> Rocksdb::isTableExist(const std::string & name) const{
  auto iter = _tables.find(name);
  if(iter != _tables.end()){ //find it
    return std::make_pair(true, iter->second);
  }
  return std::make_pair(false, (::rocksdb::ColumnFamilyHandle *)nullptr);

}
std::shared_ptr<Table> Rocksdb::getTable(const std::string & name) const{
  auto && [exist, handle] = isTableExist(name);
  if(exist){
    return std::make_shared<Table>(_db, name, handle, *_options);
  }
  return std::shared_ptr<Table>();
}
std::shared_ptr<Table> Rocksdb::createTable(const std::string & name){
  auto [exist, handle] = isTableExist(name);
  if(!exist){ // not exist 
    auto s = _db->CreateColumnFamily(_options->options(), name, &handle);
    if(!s.ok()){
      std::cerr << "create column family failed:" << s.ToString() << std::endl;
      return std::shared_ptr<Table>();
    }

    _tables.insert(std::make_pair(name, handle));

    return std::make_shared<Table>(_db, name, handle, *_options);
  }
  return std::make_shared<Table>(_db, name, handle, *_options);
}
bool Rocksdb::closeTable(const std::string & name){
  auto &&[exist, handle]  = isTableExist(name);
  if(exist){
    _tables.erase(name);
    return _closeColumnFamilyHandle(handle);
  }
  return true;
}

bool Rocksdb::dropTable(const std::string & name){
  auto &&[exist, handle]  = isTableExist(name);
  if(exist){
    _tables.erase(name);

    auto s = _db->DropColumnFamily(handle);
    if(!s.ok()){
      std::cerr << "ERROR: drop column family" << s.ToString() << std::endl;
      return false; 
    }

    return _closeColumnFamilyHandle(handle);
  }
  return true;
}

bool Rocksdb::_closeColumnFamilyHandle(::rocksdb::ColumnFamilyHandle * handle){
  assert(handle != nullptr);

  auto s =_db->DestroyColumnFamilyHandle(handle);
  if(!s.ok()){
    std::cerr << "destroy column family" << s.ToString() << std::endl;
    return false;
  }
  return true;
}

using ColumnFamilyDescriptors = ::std::vector<::rocksdb::ColumnFamilyDescriptor>;

static bool currentColumnFamilyDescriptors(const std::string &name, const ::rocksdb::Options & options, ColumnFamilyDescriptors & cfDescriptors) {
  //1. get all columns  from exist db & save the tables handle
  std::vector<std::string> cfNames;
  auto && s = ::rocksdb::DB::ListColumnFamilies(options, name, &cfNames); //DBOptions
  if(!s.ok()) {  
    std::cerr << s.ToString() << std::endl;
    return false;
  }

  //2. create ColumnFamilyDescriptors from cfNames
  for(auto && name : cfNames){
    cfDescriptors.push_back(::rocksdb::ColumnFamilyDescriptor(name, options)); //ColumnOptions
  }
    return true;
}

bool Rocksdb::_openExist(){
  ColumnFamilyDescriptors  cfDescriptors;
  if(!currentColumnFamilyDescriptors(_name, _options->options(), cfDescriptors)) 
      return false;

  //2. open the db with current ColumnFamilies , return cfHandles
  ::rocksdb::DB* rdb = nullptr;
  std::vector<::rocksdb::ColumnFamilyHandle*> cfHandles;
  auto && s = ::rocksdb::DB::Open(_options->options(), _name, cfDescriptors, &cfHandles, &rdb);
  if(!s.ok()){
    std::cerr << "open failed:" << s.ToString() << std::endl;
    return false;
  }
    //3. make db instance
  _db.reset(rdb, [this](::rocksdb::DB * rdb) {
      std::cout << this->_name << " is deleted" << std::endl;
      delete rdb; 
  });

  //4. save the handles of column family
  for(auto && handle : cfHandles){
    _tables.insert(std::make_pair(handle->GetName(), handle));
  }
  return true;
}

::rocksdb::Status Rocksdb::put(const ::rocksdb::Slice & key, const ::rocksdb::Slice & value){
  return _db->Put(_options->write(), key, value);
}
::rocksdb::Status Rocksdb::get(const ::rocksdb::Slice & key, std::string & value){
  return _db->Get(_options->read(), key, &value);
}
::rocksdb::Status Rocksdb::del(const ::rocksdb::Slice & key){
  return _db->Delete(_options->write(), key);
}
bool Rocksdb::size(size_t & sz) const{
    sz = 0;
    return foreach([&sz](const auto & , const auto & ) -> bool{
        ++sz; return true;
    });
}
bool Rocksdb::foreach(const Visitor & v, const Slice & start, const Slice & end) const {
  std::unique_ptr<::rocksdb::Iterator> iter(_db->NewIterator(_options->read()));
  if(iter == nullptr){
    std::cerr << "ERROR: new iterator null" << std::endl;
    return false; 
  }

    if(start == Slice()){
        iter->SeekToFirst();
    }
    else{
        iter->Seek(start);
    }
  for(; iter->Valid() && _options->comparator().Compare(iter->key(), end) != 0; iter->Next()){
        if(!v(iter->key(), iter->value())) return false;
  }
  return true;
}


bool Rocksdb::destroy(const std::string & name){
    ::rocksdb::Options options; //TODO:  use the options same as db-instance
  ColumnFamilyDescriptors cfDescriptors;
    if(!currentColumnFamilyDescriptors(name, options, cfDescriptors)) return false;
    return ::rocksdb::DestroyDB(name, options, cfDescriptors).ok();
}

}//end namespace rocksdb
}//end namespace helper
